package server;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.HashMap;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import server.Message.MsgType;

public class LivenessThread extends Thread {

	private static final long REQ_TIMEOUT = 1000;
	private static final long PING_TIMEOUT = 1000;

	private static final int CONFIG_PORT_OFF = 13000;

	private void log(String msg) {
		Server.log(this.getName(), msg);
	}

	private boolean reqResp = true; // got a req response
	private boolean writeConfig = false; // org mesh updated
	private String filename; // config file

	/* Network stuff */
	private DatagramChannel sock;
	private Selector sel;

	private int joinNumber = 0; // liveness seq #

	private Map<String, Integer> timesAdded;
	private Map<String, Map<SocketAddress, Integer>> addrSeen;

	private BlockingQueue<Message> connections;
	
	/* Addresses */
	private Map<String, CopyOnWriteArrayList<SocketAddress>> sAddrs;
	private ConcurrentHashMap<String, SocketAddress> uAddrs;

	public LivenessThread(String fileName, int port,
			Map<String, CopyOnWriteArrayList<SocketAddress>> sAddrs,
			ConcurrentHashMap<String, SocketAddress> uAddrs,
			Map<String, Map<SocketAddress, Integer>> addrSeen,
			BlockingQueue<Message> connections) {
		super("Live-" + port);
		this.sAddrs = sAddrs;
		this.uAddrs = uAddrs;
		this.filename = fileName;
		this.addrSeen = addrSeen;
		this.connections = connections;
		try {
			sel = Selector.open();
			sock = DatagramChannel.open();
			sock.socket().bind(new InetSocketAddress(port));
			sock.configureBlocking(false);
			sock.register(sel, SelectionKey.OP_READ);
		}
		catch (SocketException e) {
			e.printStackTrace();
		}
		catch (IOException e) {
			e.printStackTrace();
		}

	}

	@Override
	public void run() {
		sendRequests();
		//long end = 0;
		while (true) {
			try {
				/*if (end == 0)
					end = System.currentTimeMillis() + PING_TIMEOUT;

				sel.select(end - System.currentTimeMillis());*/
				sel.select(PING_TIMEOUT);
				
				if (sel.selectedKeys().size() != 0)
					readMessages();
				else {
					// write config if there are changes
					//if (writeConfig)
					//	writeConfig();
					log("Load: " + (Server.LOAD - uAddrs.size()));
					//sendPing();
					//end = 0;
				}

			}
			catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/*private void sendPing() {
		Message msg;
		try {
			msg = new Message(MsgType.PING, Server.EMPTY_ID, joinNumber, uAddrs.size() + "");

			ByteBuffer buf = msg.serialize();
			for (SocketAddress address : sAddrs.get(Server.EMPTY_ID)) {
				if (address == null)
					continue;

				InetSocketAddress ina = (InetSocketAddress) address;
				SocketAddress livenessAddress = new InetSocketAddress(ina.getAddress(),
						ina.getPort() - 1000);
				sock.send(buf, livenessAddress);
				// resets the buffer so it can be sent again
				buf.flip();
			}
		}
		catch (IOException e) {
			e.printStackTrace();
		}
	}*/

	private boolean isFromMyself(InetSocketAddress inetAddr) {
		return inetAddr.getAddress().isLoopbackAddress()
				&& inetAddr.getPort() == sock.socket().getLocalPort() + 2000;
	}

	private void consumeMessage(Message msg, SocketAddress addr) {
		switch (msg.type) {
		case HERE:
			handleHere(msg, addr);
			break;
		case REQUEST:
			handleRequest(msg, addr);
			break;
		case SURVEY:
			handleSurvey(msg, addr);
			break;
		case LOAD:
			msg.addr = addr;
			try {
				connections.put(msg);
			}
			catch (InterruptedException e) {
				e.printStackTrace();
			}
			break;
		case PING:
			break;
		}

	}

	private void handleHere(Message msg, SocketAddress addr) {
		InetSocketAddress inetAddr = (InetSocketAddress) addr;
		addrSeen.get(msg.pubId).put(inetAddr, (int) msg.seq);
		reqResp = true;

		tryToAdd(inetAddr, msg.pubId);

		// track how many times i've been added
		if (!msg.data.equals("")) {
			Integer times = timesAdded.get(msg.pubId);
			timesAdded.put(msg.pubId, (times == null ? 0 : times + 1));
		}

	}
	
	private void handleSurvey(Message msg, SocketAddress addr) {
		InetSocketAddress inetAddr = (InetSocketAddress) addr;

		// is a self-survey request
		if (isFromMyself(inetAddr)) {
			log("Self survey request");
			surveyMesh(msg);
			return;
		}
		
		String[] params = msg.data.split(" ");
		int load = Integer.parseInt(params[1]);
		String mesh;
		if ((msg.flags & 0x1) == 0) {
			if (params.length > 3) {
				inetAddr = new InetSocketAddress(params[3], Integer.valueOf(params[4]));
			}
			mesh = params[2];
		}
		else {
			if (params.length > 2) {
				inetAddr = new InetSocketAddress(params[2], Integer.valueOf(params[3]));
			}
			mesh = "";
		}
		
		// last seq # seen by addr
		Integer lastNum = addrSeen.get(msg.pubId).get(inetAddr);
		long requestNumber = msg.seq;

		// seen before / old msgs
		if (lastNum != null && lastNum >= requestNumber) {
			log("discarded SURVEY from " + inetAddr + " request " + lastNum);
			return;
		}

		addrSeen.get(msg.pubId).put(inetAddr, (int) requestNumber);
		
		log("ID: " + params[0] + ", Load: " + load + ", Mesh: " + mesh);
		boolean reply = false;
		boolean inMesh = sAddrs.get(mesh) != null;
		switch (msg.flags & 0x1) {
		case 0x0:
			log("In mesh: " + inMesh + "(" + ((msg.flags & 0x2) == 0) + ")");
			reply = reply || ((msg.flags & 0x2) == 0 && inMesh);
		default:
			reply = reply || Server.calculateLoad(uAddrs.size()) > load;
			reply = reply && Server.calculateLoad(uAddrs.size()) > 0;
			if (reply) { // i should respond
				try {
					load = Server.calculateLoad(uAddrs.size());
					log("Responding with LOAD " + load);
					Message m;
					
					// create load message (incl yes if i'm in the mesh)
					if (inMesh && (msg.flags & 0x1) == 0) {
						m = new Message(MsgType.LOAD, params[0], msg.seq, load + " yes");
					}
					else {
						m = new Message(MsgType.LOAD, params[0], msg.seq, load + "");
					}
					
					// update the load to send along
					if ((msg.flags & 0x1) == 0)
						msg.setData(params[0] + " " + load + " " + mesh);
					else
						msg.setData(params[0] + " " + load);
					
					m.flags = msg.flags;
					sock.send(m.serialize(), inetAddr);
					
				}
				catch (IOException e) {
					e.printStackTrace();
				}
			}
			break;
		}

		
		try {
			msg.setData(msg.data + " " + inetAddr.getAddress().getHostAddress() + " " + inetAddr.getPort());
		}
		catch (UnsupportedEncodingException e1) {
			e1.printStackTrace();
		}

		CopyOnWriteArrayList<SocketAddress> mAddrs;
		ByteBuffer buf = msg.serialize();
		
		// if i'm in the mesh, send only to those
		if (sAddrs.containsKey(msg.pubId))
			mAddrs = sAddrs.get(msg.pubId);
		// if i'm not, send to the org mesh
		else
			mAddrs = sAddrs.get(Server.EMPTY_ID);

		// sending to neighbors
		for (SocketAddress address : mAddrs) {
			if (address != null) {
				try {
					InetSocketAddress ina = (InetSocketAddress) address;
					SocketAddress livenessAddress = new InetSocketAddress(ina.getAddress(),
							ina.getPort() - 1000);
					if (livenessAddress.equals(inetAddr))
						continue;
					
					sock.send(buf, livenessAddress);
					buf.flip(); // resets the buffer so it can be sent again
				}
				catch (IOException e) {
					e.printStackTrace();
				}
			}
		}

	}

	private void handleRequest(Message msg, SocketAddress addr) {
		InetSocketAddress inetAddr = (InetSocketAddress) addr;

		// is a new mesh request
		if (isFromMyself(inetAddr)) {
			addrSeen.put(msg.pubId, new ConcurrentHashMap<SocketAddress, Integer>());
			joinMesh(msg.pubId);
			return;
		}

		// otherwise it's from somebody else
		if (!msg.data.equals("")) { // has addr in msg
			String[] msgsplit = msg.data.split(" ");
			inetAddr = new InetSocketAddress(msgsplit[0], Integer.valueOf(msgsplit[1]));
		}
		else { // no addr
			try {
				msg.setData(inetAddr.getAddress().getHostAddress() + " " + inetAddr.getPort());
			}
			catch (UnsupportedEncodingException e) {
				e.printStackTrace();
			}
		}

		// put the new mesh id in my table
		if (addrSeen.get(msg.pubId) == null)
			addrSeen.put(msg.pubId, new ConcurrentHashMap<SocketAddress, Integer>());

		// last seq # seen by addr
		Integer lastNum = addrSeen.get(msg.pubId).get(inetAddr);
		long requestNumber = msg.seq;

		// seen before / old msgs
		if (lastNum != null && lastNum >= requestNumber) {
			log("discarded REQ from " + inetAddr + " request " + lastNum);
			return;
		}

		ByteBuffer buf = msg.serialize();
		CopyOnWriteArrayList<SocketAddress> mAddrs;

		// if i'm in the mesh, send only to those
		if (sAddrs.containsKey(msg.pubId))
			mAddrs = sAddrs.get(msg.pubId);
		// if i'm not, send to the org mesh
		else
			mAddrs = sAddrs.get(Server.EMPTY_ID);

		// sending to neighbors
		for (SocketAddress address : mAddrs) {
			if (address != null) {
				try {
					InetSocketAddress ina = (InetSocketAddress) address;
					SocketAddress livenessAddress = new InetSocketAddress(ina.getAddress(),
							ina.getPort() - 1000);
					if (livenessAddress.equals(inetAddr))
						continue;
					sock.send(buf, livenessAddress);
					buf.flip(); // resets the buffer so it can be sent again
					log("Forwarded " + msg.toString() + " to " + livenessAddress);
				}
				catch (IOException e) {
					e.printStackTrace();
				}
			}
		}

		addrSeen.get(msg.pubId).put(inetAddr, (int) requestNumber);

		// if i'm part of the mesh, and the new guy isn't already in my list
		if (sAddrs.get(msg.pubId) != null) {
			boolean added = tryToAdd(inetAddr, msg.pubId);

			try {
				// respond to the request
				Message m;
				if (added) {
					m = new Message(MsgType.HERE, msg.pubId, requestNumber, "added");
				}
				else {
					m = new Message(MsgType.HERE, msg.pubId, requestNumber, "");
				}
				log("Sent " + m);
				sock.send(m.serialize(), inetAddr);
			}
			catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	private void readMessages() {
		ByteBuffer in = ByteBuffer.allocate(2048);
		try {
			for (SelectionKey s : sel.selectedKeys()) {
				if (s.isValid() && s.isReadable()) {
					in.clear();
					DatagramChannel ch = ((DatagramChannel) s.channel());
					SocketAddress addr = ch.receive(in);
					if (addr != null) {
						in.flip();
						Message msg = Message.deserialize(in);
						log("recieved " + msg);
						consumeMessage(msg, addr);
						sel.selectedKeys().remove(s);
					}
				}
			}
		}
		catch (IOException e) {
			log("Error!");
			e.printStackTrace();
		}
	}

	private void surveyMesh(Message m) {
		m.seq = joinNumber;
		joinNumber++;
		
		ByteBuffer buf = m.serialize();
		CopyOnWriteArrayList<SocketAddress> mAddrs = sAddrs.get(m.pubId);

		for (SocketAddress address : mAddrs) {
			if (address != null) {
				try {
					InetSocketAddress ina = (InetSocketAddress) address;
					SocketAddress livenessAddress = new InetSocketAddress(ina.getAddress(),
							ina.getPort() - 1000);
					sock.send(buf, livenessAddress);
					buf.flip(); // resets the buffer so it can be sent again
				}
				catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	private void joinMesh(String id) {
		Message msg;
		try {
			msg = new Message(MsgType.REQUEST, id, joinNumber, "");
		}
		catch (UnsupportedEncodingException e) {
			return;
		}
		joinNumber++;

		ByteBuffer buf = msg.serialize();
		CopyOnWriteArrayList<SocketAddress> mAddrs = sAddrs.get(Server.EMPTY_ID);

		for (SocketAddress address : mAddrs) {
			if (address != null) {
				try {
					InetSocketAddress ina = (InetSocketAddress) address;
					SocketAddress livenessAddress = new InetSocketAddress(ina.getAddress(),
							ina.getPort() - 1000);
					sock.send(buf, livenessAddress);
					buf.flip(); // resets the buffer so it can be sent again
				}
				catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

	private void sendRequests() {
		reqResp = false;
		//addrSeen = new HashMap<String, Map<SocketAddress, Integer>>();
		//addrSeen.put(Server.EMPTY_ID, new HashMap<SocketAddress, Integer>());
		timesAdded = new HashMap<String, Integer>();
		Scanner scan = null;
		try {
			scan = new Scanner(new File(filename));
		}
		catch (FileNotFoundException e) {
			e.printStackTrace();
		}

		Message requestMsg;
		try {
			requestMsg = new Message(MsgType.REQUEST, Server.EMPTY_ID, joinNumber, "");
		}
		catch (UnsupportedEncodingException e) {
			return;
		}
		joinNumber++;
		while (scan.hasNext()) {
			String ip = scan.next();
			int portOffset = scan.nextInt();
			int livePort = CONFIG_PORT_OFF + portOffset;

			InetSocketAddress addy = new InetSocketAddress(ip, livePort);
			long end = System.currentTimeMillis() + REQ_TIMEOUT;
			try {
				log("Sent " + requestMsg);
				sock.send(requestMsg.serialize(), addy);
				long current = System.currentTimeMillis();
				while (end > current) {

					sel.select(end - current);
					current = System.currentTimeMillis();
					readMessages();
					if (reqResp)
						return;
				}
			}
			catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	private boolean tryToAdd(InetSocketAddress addr, String meshName) {
		boolean added = false;
		SocketAddress other = new InetSocketAddress(addr.getAddress(), addr.getPort() + 1000);

		if (!sAddrs.get(meshName).contains(other)) {
			added = addIfEmpty(other, meshName);
			if (!added) {
				int index = (int) (Math.random() * addrSeen.get(meshName).size());
				if (index < Server.S) {
					sAddrs.get(meshName).set(index, other);
					added = true;
				}
			}
		}
		if (meshName.equals(Server.EMPTY_ID))
			writeConfig |= added;

		if (added) {
			log("Added: " + addr);
		}

		return added;
	}

	private boolean addIfEmpty(SocketAddress addr, String meshName) {
		CopyOnWriteArrayList<SocketAddress> l = sAddrs.get(meshName);
		int emptyIndex = Server.getEmptyIndex(l);
		if (emptyIndex >= 0) {
			l.set(emptyIndex, addr);
			return true;
		}
		return false;
	}

	private void writeConfig() {
		log("Writing org mesh to disk");
		FileWriter fstream;
		FileWriter fstream2;
		try {
			fstream = new FileWriter(filename);
			fstream2 = new FileWriter("data/" + filename);
			BufferedWriter out = new BufferedWriter(fstream);
			BufferedWriter out2 = new BufferedWriter(fstream2);
			for (SocketAddress addr : addrSeen.get(Server.EMPTY_ID).keySet()) {
				if (addr == null)
					continue;

				InetSocketAddress inetAddr = (InetSocketAddress) addr;
				out.write(String.format("%s %d\n", inetAddr.getHostName(), inetAddr.getPort()
						- CONFIG_PORT_OFF));
			}
			out.flush();
			out.close();
			fstream.close();
			for (SocketAddress addr : sAddrs.get(Server.EMPTY_ID)) {
				if (addr == null)
					continue;

				InetSocketAddress inetAddr = (InetSocketAddress) addr;
				out2.write(String.format("%s %d\n", inetAddr.getHostName(), inetAddr.getPort()
						- CONFIG_PORT_OFF - 1000));
			}
			out2.flush();
			out2.close();
			fstream2.close();
		}
		catch (IOException e) {
			log("Failed to write config file.");
		}
		writeConfig = false;
	}

	@Override
	public String toString() {
		return "Out: " + sAddrs.get(Server.EMPTY_ID);
	}
}
